package com.then.component.login.repository;


import java.io.IOException;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.greenrobot.eventbus.EventBus;

import com.luojilab.componentservice.event.ResourceLoginCompletedEvent;
import com.then.basic.socket.core.ClientRequest;
import com.then.basic.socket.core.PokerReq;
import com.then.basic.socket.core.Registration;
import com.then.basic.socket.core.error.CommonError;
import com.then.basic.socket.core.reader.ProtoStreamReader;
import com.then.basic.socket.core.reader.ReadSocketException;
import com.then.basic.socket.data.CommClasses;
import com.then.basic.socket.data.ProtocolCode;
import com.then.basic.socket.utils.I366PickUtil;
import com.then.component.login.LoginConfig;
import com.then.component.login.LoginParams;
import com.then.component.login.exception.LoginErrorResponseException;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import timber.log.Timber;

/**
 * 连接服务器的整个流程
 * Created by longyi on 2016/12/9.
 */

public class LoginAction {

    private static final String TAG = "LoginAction";

    private Disposable mDisposable;
    private List<Socket> socketToCleanUp = new ArrayList<>(2);
    private AtomicInteger loginStage = new AtomicInteger(NOT_STARTED);

    private static final int NOT_STARTED = 0;
    private static final int LOGGING_IN = 1;
    private static final int COMPLETED = 2;
    private static final int CANCELLED = 3;
    private static final int ON_ERROR = 4;

    private static class ResourceLoginAddressParams {
        InetSocketAddress socketAddress;
        String loginKey;
        int userId;
    }

    public LoginAction() {
    }

    public boolean isLoggingIn() {
        return loginStage.get() == LOGGING_IN;
    }

    public void stopLogin() {
        if (loginStage.get() != LOGGING_IN) {
            Timber.w( "stopLogin: called when not logging");
            return;
        }

        Timber.i( "stopping login");
        if (mDisposable != null) {
            mDisposable.dispose();
            mDisposable = null;
        }
        if (!loginStage.compareAndSet(NOT_STARTED, CANCELLED)) {
            loginStage.compareAndSet(LOGGING_IN, CANCELLED);
        }
        cleanUp();
    }

    public void login(LoginParams loginParams) {
        if (loginStage.get() == LOGGING_IN) {
            Timber.e( "login() called when logging in!");
            return;
        }
        Schedulers.start();
        loginStage.set(LOGGING_IN);
        Observable.just(loginParams)
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .map(loginParam -> {
                    Timber.i("connect login server");
                    Socket socket = new Socket();
                    socket.connect(new InetSocketAddress((String) LoginConfig.Login_LoginServer[0],
                            (Integer) LoginConfig.Login_LoginServer[1]), Registration.CONNECT_TIME_OUT);
                    socket.setSoTimeout(10000); // set read time out
                    socketToCleanUp.add(socket);
                    Timber.i("send login request");
                    sendLoginRequest(socket, loginParam);
                    return socket;
                }).flatMap(socket -> {
            try {
                // 读取服务器返回值，转化为资源服务器地址
                Timber.i( "read login response");
                ProtoStreamReader protoStreamReader = new ProtoStreamReader(socket.getInputStream());
                ResourceLoginAddressParams result = handleLoginServer(socket, protoStreamReader.readObj());
                if (result != null) {
                    return Observable.just(result);
                } else {
                    return Observable.error(new LoginErrorResponseException("null response of login"));
                }
            } catch (Exception e) {
                stopLogin();
                return Observable.error(e);
            } finally {
                socketToCleanUp.remove(socket);
                closeSocketSilently(socket);
            }
        }).map(resourceLoginParams -> {
            // 建立资源连接
            Timber.i( "connect res server");
            Socket socket = new Socket();
            socket.connect(resourceLoginParams.socketAddress, Registration.CONNECT_TIME_OUT);
            socket.setSoTimeout(10000); // 设置读取超时
            socketToCleanUp.add(socket);
            // 登陆资源服务器
            Timber.i( "send res login request");
            Object[][] objs = {{130, resourceLoginParams.loginKey, I366PickUtil.TYPE_STRING_UTF16}};
            byte[] bytes = I366PickUtil.packAll(objs, ProtocolCode.REQ_LOGIN_RESOURCESERVER,
                    resourceLoginParams.userId);
            OutputStream out = socket.getOutputStream();
            out.write(bytes);
            out.flush();
            return socket;
        }).flatMap(socket -> {
            ProtoStreamReader protoStreamReader = new ProtoStreamReader(socket.getInputStream());
            try {
                int status = handleResourceLogin(protoStreamReader.readObj());
                if (status == 0) { // 资源服务器登陆成功
                    return Observable.just(socket);
                } else {
                    return Observable.error(new LoginErrorResponseException("err response of login"));
                }
            } catch (Exception e) {
                // pass
                stopLogin();
                return Observable.error(e);
            }
        }).map(socket -> {
            Timber.i( "transform socket to registration");
            socketToCleanUp.remove(socket);
            return Registration.bindSocket(Registration.SERVER_TYPE_RES, socket);
        }).subscribe(new Observer<Registration>() {
            @Override
            public void onSubscribe(Disposable d) {
                mDisposable = d;
            }

            @Override
            public void onNext(Registration registration) {

            }

            @Override
            public void onError(Throwable e) {
                loginStage.set(ON_ERROR);
                cleanUp();
                if (e instanceof ConnectException) {
                    EventBus.getDefault().post(CommonError.NetworkErrorInfo.ConnectServerFailMessage);
                    Timber.e( "login error: " + e.toString());
                } else {
                    if (e instanceof ReadSocketException || e instanceof IOException
                            || e instanceof LoginErrorResponseException) {
                        EventBus.getDefault().post(new CommonError.NetworkErrorInfo(
                                CommonError.NetworkErrorInfo.ErrorType.ReadPackageFail, 0, 0));
                    }
                    Timber.e( "login error!", e);
                }
            }

            @Override
            public void onComplete() {
                loginStage.set(COMPLETED);
                cleanUp();

                if (mDisposable.isDisposed()) {
                    Timber.i( "login task return but cancelled");
                } else {
                    EventBus.getDefault().post(new ResourceLoginCompletedEvent());
                    Timber.i( "Login success!");
                }
            }
        });
    }

    private synchronized void cleanUp() {
        Schedulers.shutdown();
        cleanUpSockets();
    }

    private synchronized void cleanUpSockets() {
        for (Socket socket : socketToCleanUp) {
            closeSocketSilently(socket);
        }
        socketToCleanUp.clear();
    }

    private void closeSocketSilently(Socket socket) {
        if (socket != null) {
            if (socket.isConnected() && !socket.isClosed()) {
                try {
                    socket.close();
                } catch (IOException e) {
                    // ignore
                }
            }
        }
    }

    private void sendLoginRequest(Socket socket, LoginParams loginParam)
            throws IOException {
        PokerReq.closeResourceServer(); // 关闭资源服务器，避免被踢
        OutputStream outputStream = socket.getOutputStream();
        outputStream.write(loginParam.getRequestData());
        outputStream.flush();
    }

    private ResourceLoginAddressParams handleLoginServer(Socket socket, ClientRequest clientRequest)
            throws LoginErrorResponseException, IOException, ReadSocketException {
        if (!(clientRequest.getBody() instanceof CommClasses.LoginLoginServerInfo)) {
            throw new LoginErrorResponseException("Error response:" + clientRequest.getRequestCode()
                    + " require:" + ProtocolCode.REQ_LOGIN_LOGINSERVER);
        }

        CommClasses.LoginLoginServerInfo info = (CommClasses.LoginLoginServerInfo) clientRequest.getBody();
        if (info == null) {
            return null;
        }

        EventBus.getDefault().post(info);
        int status = info.success;
        Timber.i( "LoginLoginServer status = " + status);
        if (status == 0) { // 登录成功
            ResourceLoginAddressParams loginParams = new ResourceLoginAddressParams();
            loginParams.loginKey = info.key;
            loginParams.socketAddress = new InetSocketAddress(info.resServerIp, info.resServerPort);
            loginParams.userId = info.userId;
            return loginParams;
        } else if (status == 4) { // 需要升级
            Timber.i( "handleLoginServer: version check");
            socket.getOutputStream().write(PokerReq.getCheckVersionReq());
            CommClasses.CheckVersionData checkVersionData =
                    (CommClasses.CheckVersionData) new ProtoStreamReader(socket.getInputStream()).readObj().getBody();
            if (checkVersionData != null) {
                Timber.i( "handleLoginServer: broadcast check version data");
                EventBus.getDefault().post(checkVersionData);
            }
        }
        return null;
    }

    private int handleResourceLogin(ClientRequest clientRequest) throws LoginErrorResponseException {
        if (!(clientRequest.getBody() instanceof CommClasses.LoginResourceServerInfo)) {
            throw new LoginErrorResponseException("Error response:" + clientRequest.getRequestCode()
                    + " require:" + ProtocolCode.REQ_LOGIN_LOGINSERVER);
        }

        CommClasses.LoginResourceServerInfo info = (CommClasses.LoginResourceServerInfo) clientRequest.getBody();
        EventBus.getDefault().post(info);
        return info.success;
    }
}
